package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.app.func.DimAsyncFunction;
import realtime.beans.VideoPlayChapterBean;
import realtime.util.DateFormatUtil;
import realtime.util.DorisUtil;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Create 2023/9/9 13:51
 * 播放域-章节播放统计
 */
public class DwsPlayChapterPlayWindow {
    public static void main(String[] args) {

        //TODO 1、设置环境
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);

        DataStreamSource<String> sourceDS = env.fromSource(KafkaUtil.getKafkaConsumer("dwd_traffic_video_log", "dws_PCPW"),
                WatermarkStrategy.noWatermarks(), "dwd_traffic_video_log"
        );

        //TODO 2、etl
        SingleOutputStreamOperator<JSONObject> JsonObjDS = sourceDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    JSONObject appVideo = jsonObject.getJSONObject("appVideo");
                    String uid = jsonObject.getJSONObject("common").getString("uid");
                    Long ts = jsonObject.getLong("ts");

                    appVideo.put("uid",uid);
                    appVideo.put("ts",ts);

                    out.collect(appVideo);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner((value,ts)->value.getLong("ts")));

        //TODO 3、按章节和uid分组聚合转换
        SingleOutputStreamOperator<VideoPlayChapterBean> processDS = JsonObjDS.keyBy(value -> value.getString("video_id") + "-" + value.getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, VideoPlayChapterBean>() {

                    ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //设置状态的生存时间
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) //更新类型
                                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) //状态可见性(状态过期之后是否可读)
                                .build();
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("state", String.class);
                        valueStateDescriptor.enableTimeToLive(ttlConfig);

                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, VideoPlayChapterBean>.Context ctx, Collector<VideoPlayChapterBean> out) throws Exception {
                        VideoPlayChapterBean playChapterBean = new VideoPlayChapterBean();


                        String videoId = value.getString("video_id");
                        String uid = value.getString("uid");
                        Long ts = value.getLong("ts");
                        Long playSec = value.getLong("play_sec");

                        playChapterBean.setVideoId(videoId);
                        playChapterBean.setUserId(uid);
                        playChapterBean.setDurTime(playSec);

                        if (playSec < 30) {
                            playChapterBean.setVideoPlayCt(1L);
                        } else {
                            playChapterBean.setVideoPlayCt(0L);
                        }

                        if (valueState.value() == null) {
                            valueState.update(DateFormatUtil.toDate(ts));
                            playChapterBean.setUserCt(1L);
                        } else if (!valueState.value().equals(DateFormatUtil.toDate(ts))) {
                            valueState.update(DateFormatUtil.toDate(ts));
                            playChapterBean.setUserCt(1L);
                        }else {
                            playChapterBean.setUserCt(0L);
                        }

                        out.collect(playChapterBean);

                    }
                });

        //TODO 4、按章节(video_id)聚合,统计数据
        SingleOutputStreamOperator<VideoPlayChapterBean> reduceDS = processDS.keyBy(value -> value.getVideoId())
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)))
                .reduce(new ReduceFunction<VideoPlayChapterBean>() {
                    @Override
                    public VideoPlayChapterBean reduce(VideoPlayChapterBean value1, VideoPlayChapterBean value2) throws Exception {
                        value1.setUserCt(value1.getUserCt() + value2.getUserCt());
                        value1.setVideoPlayCt(value1.getVideoPlayCt() + value2.getVideoPlayCt());
                        value1.setDurTime(value1.getDurTime() + value2.getDurTime());
                        return value1;
                    }
                }, new WindowFunction<VideoPlayChapterBean, VideoPlayChapterBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<VideoPlayChapterBean> input, Collector<VideoPlayChapterBean> out) throws Exception {
                        VideoPlayChapterBean playChapterBean = input.iterator().next();
                        long start = window.getStart();
                        long end = window.getEnd();
                        long timeMillis = System.currentTimeMillis();

                        playChapterBean.setStt(DateFormatUtil.toYmdHms(start));
                        playChapterBean.setEdt(DateFormatUtil.toYmdHms(end));
                        playChapterBean.setCurDate(DateFormatUtil.toDate(timeMillis));
                        out.collect(playChapterBean);
                    }
                });

        //reduceDS.print();

        reduceDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction< VideoPlayChapterBean>("dim_video_info") {
            @Override
            public String getKey(VideoPlayChapterBean bean) {
                return bean.getVideoId();
            }

            @Override
            public void join(VideoPlayChapterBean bean, JSONObject dim) {
                String videoName = dim.getString("video_name");
                String chapterId = dim.getString("chapter_id");
                bean.setChapterId(chapterId);
                bean.setVideoName(videoName);
            }
        }, 60, TimeUnit.SECONDS);

        reduceDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction< VideoPlayChapterBean>("dim_chapter_info") {
            @Override
            public String getKey(VideoPlayChapterBean bean) {
                return bean.getChapterId();
            }

            @Override
            public void join(VideoPlayChapterBean bean, JSONObject dim) {
                String chapterName = dim.getString("chapter_name");
                bean.setChapterName(chapterName);
            }
        }, 60, TimeUnit.SECONDS);

        //reduceDS.print();


        SingleOutputStreamOperator<String> endDS = reduceDS.map(value -> {
            SerializeConfig serializeConfig = new SerializeConfig();
            serializeConfig.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);
            String jsonString = JSONObject.toJSONString(value,serializeConfig);
            JSONObject jsonObject = JSON.parseObject(jsonString);
            jsonObject.remove("user_id");
            jsonObject.remove("video_id");
            jsonObject.remove("video_name");
            return jsonObject.toJSONString();
        });
        //endDS.print();

        //TODO 14、数据写入doris中
        DorisSink<String> dorisSink = DorisUtil.getDorisSink("dws_play_chapter_play_window");

        endDS.sinkTo(dorisSink);



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }
}
